package org.example.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.example.model.User;
import org.example.util.JSONUtils;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
public class EventConsumer {

    // 采用监听的方式接收事件 (消息、数据)
    /*@KafkaListener(topics = "helloTopic", groupId = "helloGroup")
    public void onEvent(@Payload String event,
                        @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                        ConsumerRecord<String, String> record) {
        System.out.println("读取到的事件：" + event + ", topic : " + topic + ", partition: " + partition);
        System.out.println("读取到的事件：" + record.toString());
    }*/

    /*@KafkaListener(topics = "helloTopic", groupId = "helloGroup")
    public void onEvent(String userJSON,
                        @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                        ConsumerRecord<String, String> record) throws JsonProcessingException {
        User user = JSONUtils.toBean(userJSON, User.class);
        System.out.println("读取到的事件：" + user + ", topic : " + topic + ", partition: " + partition);
        System.out.println("读取到的事件：" + record.toString());
    }*/

    /*@KafkaListener(topics = "${kafka.topic.name}", groupId = "${kafka.consumer.group}")
    public void onEvent(String userJSON,
                        @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                        ConsumerRecord<String, String> record) throws JsonProcessingException {
        User user = JSONUtils.toBean(userJSON, User.class);
        System.out.println("读取到的事件：" + user + ", topic : " + topic + ", partition: " + partition);
        System.out.println("读取到的事件：" + record.toString());
    }*/

    /*@KafkaListener(topics = "${kafka.topic.name}", groupId = "${kafka.consumer.group}")
    public void onEvent(String userJSON,
                        @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                        ConsumerRecord<String, String> record,
                        Acknowledgment ack) {
        try {
            // 收到消息后，处理业务
            User user = JSONUtils.toBean(userJSON, User.class);
            System.out.println("读取到的事件：" + user + ", topic : " + topic + ", partition: " + partition);
            System.out.println("读取到的事件：" + record.toString());

            // 业务处理完成，给kafka服务器确认
            ack.acknowledge(); // 手动确认消息，告诉kafka服务器，该消息我已经收到了，默认情况下kafka是自动确认
        } catch (Exception e) {
            e.printStackTrace();
        }
    }*/

    @KafkaListener(groupId = "${kafka.consumer.group}", topicPartitions = {
            @TopicPartition(
                    topic = "${kafka.topic.name}",
                    partitions = {"0", "1", "2"},
                    partitionOffsets = {
                            @PartitionOffset(partition = "3", initialOffset = "3"),
                            @PartitionOffset(partition = "4", initialOffset = "3")
                    })
    })
    public void onEvent(String userJSON,
                        @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                        ConsumerRecord<String, String> record,
                        Acknowledgment ack) {
        try {
            // 收到消息后，处理业务
            User user = JSONUtils.toBean(userJSON, User.class);
            System.out.println("读取到的事件：" + user + ", topic : " + topic + ", partition: " + partition);

            // 业务处理完成，给kafka服务器确认
            ack.acknowledge(); // 手动确认消息，告诉kafka服务器，该消息我已经收到了，默认情况下kafka是自动确认
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
